package com.atguigu.utils;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.io.IOException;
import java.util.Properties;

/**
 * ClassName: MyKafkaUtil
 * Package: com.atguigu.utils
 * Description:
 *
 * @Author 尚硅谷-李增亮
 * @Create 2023/5/15 14:46
 * @Version 1.0
 */
public class MyKafkaUtil {

    //  设置主机的属性
    private static final String BOOTSTRAP_SERVER = "hadoop102:9092";

    //  创建getKafkaSource方法从kafka中读取主题数据
    public static KafkaSource<String> getKafkaSource(String topic ,String groupId){

        return  KafkaSource.<String>builder()
                .setBootstrapServers(BOOTSTRAP_SERVER)
                .setGroupId(groupId)
                .setTopics(topic)
                .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                    @Override
                    public String deserialize(byte[] message) throws IOException {
                        //  自定义反序列化器过滤出null值: 在乐left join生成的数据集中、产生撤回流的数据zaikafka中会以null值存在、需要手动过滤
                        if (message != null) {
                            return new String(message);
                        } else {
                            return null;
                        }
                    }

                    //  除了反序列化器需要重新构建、其余方法重写可以借鉴其他的反序列化器
                    @Override
                    public boolean isEndOfStream(String nextElement) {
                        return false;
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return BasicTypeInfo.STRING_TYPE_INFO;
                    }
                })
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();
    }


    //  泛型方法
    public static <T> FlinkKafkaProducer<T> getFlinkKafkaProducer(KafkaSerializationSchema<T> kafkaSerializationSchema) {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        return new FlinkKafkaProducer<T>(
                "dwd_default_topic",
                kafkaSerializationSchema,
                properties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
        //  要想保证精准一次，需要开启检查点，否则默认none
    }


    public static String getTopicDbDDL(String groupId){
        return "create table topic_db(\n" +
                "`database` string,\n" +
                "`table` string,\n" +
                "`type` string,\n" +
                "`ts` string,\n" +
                "`data` map<string,string>,\n" +
                "`old` map<string,string>\n" +
                ")"+getKafkaSourceConnOption("topic_db",groupId);

    }

    public static String getTopicDbDDLWithRT(String groupId,String timeType){
        return "create table topic_db(\n" +
                "`database` string,\n" +
                "`table` string,\n" +
                "`type` string,\n" +
                "`ts` string,\n" +
                "`data` map<string,string>,\n" +
                "`old` map<string,string>,\n" +
                "rt as to_timestamp("+timeType+"),\n" +
                "watermark for rt as rt - interval '2' second)"+getKafkaSourceConnOption("topic_db",groupId);

    }
    public static String getKafkaSourceConnOption(String topic,String groupId){
        return "with(\n" +
                "'connector' ='kafka',\n" +
                "'topic'='"+topic+"',\n" +
                "'properties.bootstrap.servers'='"+BOOTSTRAP_SERVER+"',\n" +
                "'properties.group.id'='"+groupId+"',\n" +
                "'scan.startup.mode'='latest-offset',\n" +
                "'format'='json'\n" +
                ")";

    }
    public static String getKafkaSinkConnOption(String topic){
        return "with(\n" +
                "'connector'='kafka',\n" +
                "'topic'='"+topic+"',\n" +
                "'properties.bootstrap.servers'='"+BOOTSTRAP_SERVER+"',\n" +
                "'format'='json')";
    }

    public static FlinkKafkaProducer<String> getFlinkKafkaProducer(String topic){
        return new FlinkKafkaProducer<String>(BOOTSTRAP_SERVER,topic,new SimpleStringSchema());
    }



}
